package com.zyx.flinkdemo.stream.window.trigger;

import com.zyx.flinkdemo.pojo.WordTs;
import com.zyx.flinkdemo.stream.utils.TimeUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author zyx
 * @since 2021/6/21 20:51
 * desc: 事件时间 + 处理时间超时 触发的窗口计算
 */
public class EventTimeoutTriggerDemo {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WordTs> wordTsDs = env
                .socketTextStream("bd001", 7777)
                .map(new MapFunction<String, WordTs>() {
                    @Override
                    public WordTs map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WordTs(split[0], Long.parseLong(split[1]));
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<WordTs>forMonotonousTimestamps()
                        .withTimestampAssigner((SerializableTimestampAssigner<WordTs>)
                                (element, recordTimestamp) -> element.getTs()));

        SingleOutputStreamOperator<String> out = wordTsDs
                .keyBy(WordTs::getWord)
                .window(TumblingEventTimeWindows.of(Time.minutes(1L)))
                .trigger(EventTimeoutTrigger.of(Time.seconds(15L)))
                .aggregate(
                        // 累加器
                        new AggregateFunction<WordTs, Integer, Integer>() {
                            @Override
                            public Integer createAccumulator() {
                                return 0;
                            }

                            @Override
                            public Integer add(WordTs value, Integer accumulator) {
                                return accumulator + 1;
                            }

                            @Override
                            public Integer getResult(Integer accumulator) {
                                return accumulator;
                            }

                            @Override
                            public Integer merge(Integer a, Integer b) {
                                return a + b;
                            }
                        },
                        new RichWindowFunction<Integer, String, String, TimeWindow>() {

                            private MapState<String, Integer> mapState;

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                mapState = getRuntimeContext()
                                        .getMapState(new MapStateDescriptor<String, Integer>(
                                                "map_state",
                                                Types.STRING,
                                                Types.INT
                                        ));
                            }

                            @Override
                            public void apply(String key,
                                              TimeWindow window,
                                              Iterable<Integer> input,
                                              Collector<String> out) throws Exception {
                                String windowEndStr = TimeUtils.getFormatTimeStr(window.getEnd());
                                String mapKey = key + ":" + windowEndStr;
                                Integer lastCount = mapState.get(mapKey);
                                if (lastCount == null) {
                                    lastCount = 0;
                                }
                                Integer count = input.iterator().next();
                                // 更新数据则输出结果到流中, 否则打印由于超时引发计算
                                if (!count.equals(lastCount)) {
                                    out.collect("+++++++++++++ update trigger >>> word :" + key + ", count :" + count
                                    + ", windowEnd:" + mapKey);
                                } else {
                                    out.collect("------------- timeout trigger...");
                                }
                                // 更新mapstate
                                mapState.put(mapKey, count);
                            }
                        });

        env.execute("event_timeout_trigger");
    }
}
